package phase03.module04.code.assignment2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 第三阶段模块四作业第二题
 *
 * @author Alex Zhang
 */
public class HbaseOps {
    private static Connection connection = null;

    private static void init() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "node1,node2");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void destroy() {
        if (null != connection) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private static byte[] string2Bytes(String s) {
        return Bytes.toBytes(s);
    }

    private static void createTable(String tableName, String... families) {
        HBaseAdmin admin = null;
        try {
            admin = (HBaseAdmin) connection.getAdmin();
            HTableDescriptor relations = new HTableDescriptor(TableName.valueOf(tableName));
            for (String family : families) {
                relations.addFamily(new HColumnDescriptor(family));
            }
            admin.createTable(relations);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static void putData(String tableName, String family, String rowId, String... values) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(string2Bytes(rowId));
            for (String friend : values) {
                put.addColumn(string2Bytes(family), string2Bytes(friend), string2Bytes(friend));
            }
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 根据row key删除数据
     *
     * @param tableName 表名
     * @param rowId     row key
     */
    private static void deleteData(String tableName, String rowId) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(string2Bytes(rowId));
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 删除关系
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowId     row key
     * @param friendIds 解除关系的id
     */
    private static void unfriend(String tableName, String family, String rowId, String... friendIds) {
        Table table = null;
        List<Delete> deletes = new ArrayList<>();
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(string2Bytes(rowId));
            for (String friendId : friendIds) {
                // 删除关系
                delete.addColumn(string2Bytes(family), string2Bytes(friendId));
                // 回删关系
                Delete deleteBack = new Delete(string2Bytes(friendId));
                deleteBack.addColumn(string2Bytes(family), string2Bytes(rowId));
                deletes.add(deleteBack);
            }
            deletes.add(delete);
            table.delete(deletes);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static void displayData(String tableName) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                    String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        init();
        String tableName = "relations";
        String family = "friends";
        createTable(tableName, family);
        putData(tableName, family, "uid1", "uid2", "uid5", "uid7");
        putData(tableName, family, "uid2", "uid1", "uid3", "uid6", "uid10");
        putData(tableName, family, "uid3", "uid4");
        putData(tableName, family, "uid4", "uid3", "uid5");
        putData(tableName, family, "uid5", "uid4");
        displayData(tableName);
        unfriend(tableName, family, "uid1", "uid2");
        System.out.println("====================================");
        displayData(tableName);
        destroy();
    }

}
